package org.jaychan.live.user.provider.service.impl;

import com.alibaba.fastjson.JSON;
import com.google.common.collect.Maps;
import jakarta.annotation.Resource;
import org.apache.rocketmq.client.producer.MQProducer;
import org.apache.rocketmq.common.message.Message;
import org.jaychan.live.common.interfaces.topic.UserProviderTopicNames;
import org.jaychan.live.common.interfaces.utils.ConvertBeanUtils;
import org.jaychan.live.framework.redis.key.UserProviderCacheKeyBuilder;
import org.jaychan.live.user.constants.CacheAsyncDeleteCode;
import org.jaychan.live.user.dto.UserCacheAsyncDeleteDTO;
import org.jaychan.live.user.dto.UserDTO;
import org.jaychan.live.user.provider.dao.mapper.IUserMapper;
import org.jaychan.live.user.provider.dao.po.UserPO;
import org.jaychan.live.user.provider.service.IUserService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.core.RedisOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.SessionCallback;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.util.*;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;


@Service
public class UserServiceImpl implements IUserService {


    private static Logger LOGGER = LoggerFactory.getLogger(UserServiceImpl.class);


    @Resource
    private IUserMapper userMapper;
    @Resource
    private RedisTemplate<String, UserDTO> redisTemplate;
    @Resource
    private UserProviderCacheKeyBuilder userProviderCacheKeyBuilder;
    @Resource
    private MQProducer mqProducer;
    @Resource
    private UserProviderCacheKeyBuilder cacheKeyBuilder;


    @Override
    public UserDTO getByUserid(Long userId) {

        if(userId == null){
            return null;
        }
        String key = userProviderCacheKeyBuilder.buildUserInfoKey(userId);
        UserDTO userDTO = redisTemplate.opsForValue().get(key);

        if(userDTO != null){
            LOGGER.info("userDto is :{}",userDTO);
            return userDTO;
        }

        userDTO = ConvertBeanUtils.convert(userMapper.selectById(userId),UserDTO.class);
        if(userId != null){
            redisTemplate.opsForValue().set(key,userDTO,30, TimeUnit.MINUTES);
        }

        LOGGER.info("userDto is :{}",userDTO);
        return userDTO;
    }

    @Override
    public boolean updateUserInfo(UserDTO userDTO) {
        if (userDTO == null || userDTO.getUserId() == null) {
            return false;
        }
        int updateStatus = userMapper.updateById(ConvertBeanUtils.convert(userDTO, UserPO.class));
        if (updateStatus > -1) {
            String key = cacheKeyBuilder.buildUserInfoKey(userDTO.getUserId());
            redisTemplate.delete(key);
            UserCacheAsyncDeleteDTO userCacheAsyncDeleteDTO = new UserCacheAsyncDeleteDTO();
            userCacheAsyncDeleteDTO.setCode(CacheAsyncDeleteCode.USER_INFO_DELETE.getCode());
            Map<String,Object> jsonParam = new HashMap<>();
            jsonParam.put("userId",userDTO.getUserId());
            userCacheAsyncDeleteDTO.setJson(JSON.toJSONString(jsonParam));
            Message message = new Message();
            message.setTopic(UserProviderTopicNames.CACHE_ASYNC_DELETE_TOPIC);
            message.setBody(JSON.toJSONString(userCacheAsyncDeleteDTO).getBytes());
            //延迟一秒进行缓存的二次删除
            message.setDelayTimeLevel(1);
            try {
                mqProducer.send(message);
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }
        return true;
    }

    @Override
    public boolean insertOne(UserDTO userDTO) {
        if(userDTO == null|| userDTO.getUserId() == null){
            return false;
        }
        UserPO userPO = ConvertBeanUtils.convert(userDTO, UserPO.class);
        return userMapper.insert(userPO) == 1;
    }

    @Override
    public Map<Long, UserDTO> batchQueryUserInfo(List<Long> userIds) {
        if (CollectionUtils.isEmpty(userIds)) {
            return Maps.newHashMap();
        }
        userIds = userIds.stream().filter(id -> id > 1000).collect(Collectors.toList());
        if (CollectionUtils.isEmpty(userIds)) {
            return Maps.newHashMap();
        }

        //性能不好，片键如果差距太大，会分批查询(union all)
        //userMapper.selectBatchIds(userIds);

        //redis
//        userIds.forEach(userId->{
//            //不推荐
////            redisTemplate.opsForValue()
//        });

        List<String> keyList = new ArrayList<>();
        userIds.forEach(userId -> {
            keyList.add(userProviderCacheKeyBuilder.buildUserInfoKey(userId));
        });

        List<UserDTO> userDTOS = redisTemplate.opsForValue().multiGet(keyList).stream().filter(Objects::nonNull).toList();
        if (!CollectionUtils.isEmpty(userDTOS) && userIds.size() == userDTOS.size()) {
            return userDTOS.stream().collect(Collectors.toMap(UserDTO::getUserId, userDTO -> userDTO));
        }

        List<Long> userIdInCacheList = userDTOS.stream().map(UserDTO::getUserId).toList();
        List<Long> userIdNotInCacheList = userIds.stream().filter(userId -> !userIdInCacheList.contains(userId)).toList();

        //多线程查询替换了原来的unionAll，对java性能有要求
        Map<Long, List<Long>> userIdMap = userIdNotInCacheList.stream().collect(Collectors.groupingBy(userId -> userId / 100));
        //因为这里是用到多线程查询，往里面赛数据的时候担心涉及到数据的扩容和安全性，用了copyOnWriteArrayList
        List<UserDTO> dbQueryResult = new CopyOnWriteArrayList<>();
        userIdMap.values().parallelStream().forEach(queryUserIdList->{
            dbQueryResult.addAll(ConvertBeanUtils.convertList(userMapper.selectBatchIds(queryUserIdList),UserDTO.class));
        });

        //要不要塞入缓存？结合业务场景分析有没有必要
        if (!CollectionUtils.isEmpty(dbQueryResult)) {
            Map<String, UserDTO> saveCacheMap = dbQueryResult.stream().collect(Collectors.toMap(p -> userProviderCacheKeyBuilder.buildUserInfoKey(p.getUserId()), p -> p));
            redisTemplate.opsForValue().multiSet(saveCacheMap);

            //通过管道的形式实现批量命令功能,对网络开销较小
            redisTemplate.executePipelined(new SessionCallback<Object>() {
                @Override
                public <K, V> Object execute(RedisOperations<K, V> operations) throws DataAccessException {
                    for (String key : saveCacheMap.keySet()) {
                        operations.expire((K) key,createRandomExpireTime(),TimeUnit.SECONDS);
                    }
                    return null;
                }
            });
            if(CollectionUtils.isEmpty(userDTOS)){
                userDTOS = new ArrayList<>();
            }

            userDTOS.addAll(dbQueryResult);
        }

        return userDTOS.stream().collect(Collectors.toMap(UserDTO::getUserId,userDTO -> userDTO));
    }

    private long createRandomExpireTime(){
        return ThreadLocalRandom.current().nextInt(1000) + 60 * 30;
    }
}
